package com.ds.infrastructure.hyperspace.container.flow;

import com.ds.infrastructure.hyperspace.common.EndpointNode;
import com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig;
import com.ds.infrastructure.hyperspace.common.flow.FlowHandlerPipeline;
import com.ds.infrastructure.hyperspace.common.flow.FlowHandlerWrapper;
import com.ds.infrastructure.hyperspace.common.flow.TargetNodeWrapper;
import com.ds.infrastructure.hyperspace.common.flow.handler.LoadBalancePreFlowHandler;
import com.ds.infrastructure.hyperspace.common.flow.handler.PostFlowHandler;
import com.ds.infrastructure.hyperspace.common.flow.handler.PreFlowHandler;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.NettyTargetEndpoint;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.handler.loadbalance.RoundRobinPreHandler;
import com.ds.infrastructure.hyperspace.container.flow.operation.EntryFlowOperation;
import com.ds.infrastructure.hyperspace.container.flow.operation.TargetFlowOperation;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;

/**
 * @author  Weishaoying
 */
@Slf4j
public class HyperspaceDuplexFlowPipeline implements FlowHandlerPipeline<HyperspaceDuplexFlowContext> {

    private final EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation;

    private final TargetFlowOperation<HyperspaceDuplexFlowContext> targetFlowOperation;

    private final FlowPipelineOrchestrator pipelineOrchestrator;

    @Getter
    private final NettyTargetEndpoint targetEndpoint;

    private volatile LoadBalancePreFlowHandler<HyperspaceDuplexFlowContext> loadbalanceFlowHandler;

    private volatile FlowHandlerWrapper<PreFlowHandler<HyperspaceDuplexFlowContext>> firstPreFlowHandlerWrapper;

    private volatile FlowHandlerWrapper<PreFlowHandler<HyperspaceDuplexFlowContext>> currentPreFlowHandlerWrapper;

    private volatile FlowHandlerWrapper<PostFlowHandler<HyperspaceDuplexFlowContext>> firstPostFlowHandlerWrapper;

    private volatile FlowHandlerWrapper<PostFlowHandler<HyperspaceDuplexFlowContext>> currentPostFlowHandlerWrapper;

    public HyperspaceDuplexFlowPipeline(FlowPipelineOrchestrator pipelineOrchestrator,
                                        EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation,
                                        TargetFlowOperation<HyperspaceDuplexFlowContext> targetFlowOperation,
                                        NettyTargetEndpoint targetEndpoint) {
        this.entryFlowOperation = entryFlowOperation;
        this.targetFlowOperation = targetFlowOperation;
        this.pipelineOrchestrator = pipelineOrchestrator;
        this.targetEndpoint = targetEndpoint;
    }

    @Override
    public String getTargetServiceName() {
        return targetEndpoint.getNamespace();
    }

    @Override
    public void registerPipelineHost(String hostName) {
        pipelineOrchestrator.registerPipelineHost(hostName, this);
    }

    public boolean matchAndHandleInput(HyperspaceConnector hyperspaceConnector,
                                       HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        firstPreFlowHandlerWrapper.getFlowHandler().handleInput(duplexFlowContext, data);
        if (duplexFlowContext.isRouteMatch()) {
            duplexFlowContext.setFlowHandlerPipeline(this);
            duplexFlowContext.setTargetEndpoint(targetEndpoint);
            handleInput(hyperspaceConnector, duplexFlowContext, data);
            return true;
        }
        return false;
    }

    public void handleInput(HyperspaceConnector hyperspaceConnector,
                            HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        FlowHandlerWrapper<PreFlowHandler<HyperspaceDuplexFlowContext>> nextFlowHandlerWrapper =
                firstPreFlowHandlerWrapper.getNextFlowHandlerWrapper();
        while (nextFlowHandlerWrapper != null) {
            nextFlowHandlerWrapper.getFlowHandler().handleInput(duplexFlowContext, data);
            nextFlowHandlerWrapper = nextFlowHandlerWrapper.getNextFlowHandlerWrapper();
            if (duplexFlowContext.isClosed()) {
                return;
            }
        }
        hyperspaceConnector.handleInput(this, targetEndpoint, duplexFlowContext, data);
    }

    @Override
    public void handleOutput(HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        if (duplexFlowContext.isClosed()) {
            return;
        }
        FlowHandlerWrapper<PostFlowHandler<HyperspaceDuplexFlowContext>> nextFlowHandlerWrapper =
                firstPostFlowHandlerWrapper;
        while (nextFlowHandlerWrapper != null) {
            nextFlowHandlerWrapper.getFlowHandler().handleOutput(duplexFlowContext, data);
            nextFlowHandlerWrapper = nextFlowHandlerWrapper.getNextFlowHandlerWrapper();
            if (duplexFlowContext.isClosed()) {
                return;
            }
        }
        entryFlowOperation.handleOutput(duplexFlowContext, data);
    }

    @Override
    public EndpointNode<TargetNodeConfig> selectNode(HyperspaceDuplexFlowContext duplexFlowContext,
                                                     Map<String, ? extends TargetNodeWrapper> nodeContextMap) {
        duplexFlowContext.touchForConnection();
        return loadbalanceFlowHandler.select(duplexFlowContext, targetEndpoint.getEndpointNodes(
                duplexFlowContext.getZoneId()), nodeContextMap);
    }

    public void setNextPreFlowHandler(PreFlowHandler<HyperspaceDuplexFlowContext> flowHandler) {
        if (flowHandler instanceof LoadBalancePreFlowHandler) {
            if (loadbalanceFlowHandler != null) {
                throw new FlowHandlerInitException("Flow handler init failed," +
                        "loadbalance flow handler duplicate!");
            }

            loadbalanceFlowHandler = (LoadBalancePreFlowHandler<HyperspaceDuplexFlowContext>) flowHandler;
            return;
        }

        FlowHandlerWrapper<PreFlowHandler<HyperspaceDuplexFlowContext>> tempPreFlowHandlerWrapper =
                new FlowHandlerWrapper(flowHandler);
        if (firstPreFlowHandlerWrapper == null) {
            firstPreFlowHandlerWrapper = tempPreFlowHandlerWrapper;
        }
        if (currentPreFlowHandlerWrapper != null) {
            currentPreFlowHandlerWrapper.setNextFlowHandlerWrapper(tempPreFlowHandlerWrapper);
        }
        currentPreFlowHandlerWrapper = tempPreFlowHandlerWrapper;
    }

    public void setNextPostFlowHandler(PostFlowHandler<HyperspaceDuplexFlowContext> flowHandler) {
        FlowHandlerWrapper<PostFlowHandler<HyperspaceDuplexFlowContext>> tempPostFlowHandlerWrapper =
                new FlowHandlerWrapper<>(flowHandler);
        if (firstPostFlowHandlerWrapper == null) {
            firstPostFlowHandlerWrapper = tempPostFlowHandlerWrapper;
        }
        if (currentPostFlowHandlerWrapper != null) {
            currentPostFlowHandlerWrapper.setNextFlowHandlerWrapper(tempPostFlowHandlerWrapper);
        }
        currentPostFlowHandlerWrapper = tempPostFlowHandlerWrapper;
    }

    public void optimize() {
        if (loadbalanceFlowHandler == null) {
            loadbalanceFlowHandler = new RoundRobinPreHandler();
        }
    }
}
